package com.lhd.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lhd.app.func.ProvinceFunction;
import com.lhd.app.func.SplitAgeFunction;
import com.lhd.app.func.SplitTaqiFunction;
import com.lhd.bean.TaqiUserInfo; // 确保有这个POJO类
import com.lhd.utils.MyKafkaUtil;
import com.lhd.utils.MyClickHouseUtil;
import com.lhd.utils.MyKafkaUtil2;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class DwsTaqiUserInfo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 从Kafka读取数据
        DataStreamSource<String> kafkaStream = env.addSource(
                MyKafkaUtil2.getFlinkKafkaConsumer("dwd_taqi_user_info", "taqi_user_info_consumer")
        );

        // 解析JSON，提取after字段并转换为TaqiUserInfo对象
        DataStream<TaqiUserInfo> userInfoStream = kafkaStream.map(new MapFunction<String, TaqiUserInfo>() {
            @Override
            public TaqiUserInfo map(String value) throws Exception {
                JSONObject jsonObj = JSON.parseObject(value);
                jsonObj.getJSONObject("after").put("start_time","2025-09-19 19:31:00");
                jsonObj.getJSONObject("after").put("end_time","2025-09-19 19:32:00");
                int age = jsonObj.getJSONObject("after").getInteger("age");
                String age_group = new SplitAgeFunction().splitAge(age);
                jsonObj.getJSONObject("after").put("age_group",age_group);
                int taqiValue = jsonObj.getJSONObject("after").getInteger("taqi_value");
                String taqi_group = new SplitTaqiFunction().splitTaqi(taqiValue);
                jsonObj.getJSONObject("after").put("taqi_group",taqi_group);
                int province_id = jsonObj.getJSONObject("after").getInteger("province_id");
                String province = new ProvinceFunction().getProvince(province_id);
                jsonObj.getJSONObject("after").put("province",province);
                String afterStr = jsonObj.getString("after");
                return JSON.parseObject(afterStr, TaqiUserInfo.class);
            }
        });


        // 使用工具类写入ClickHouse
        userInfoStream.print();



        userInfoStream.addSink(MyClickHouseUtil.getSinkFunction("INSERT INTO taqi_user_info VALUES (?,?,?,?,?,?,?,?,?)"));

        env.execute("DwsTaqiUserInfo");
    }
}